package com.gome.ocean.service.zookeeper.curator;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.I0Itec.zkclient.exception.ZkException;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.gome.ocean.common.constants.ZkContants;
import com.gome.ocean.common.utils.AddressUtils;
import com.gome.ocean.service.zookeeper.client.AbstractZkClient;
import com.gome.ocean.service.zookeeper.listener.ChildListener;
import com.gome.ocean.service.zookeeper.listener.DataListener;
import com.gome.ocean.service.zookeeper.listener.StateListener;

/**
 * Curator zk 客户端实现
 * 
 * @author haqiaolong 2016年7月13日 下午6:33:42
 * @since 5.0.0
 */
public class CuratorZkClient extends AbstractZkClient<CuratorZkClient.PathChildrenListener, CuratorZkClient.NodeListener> implements InitializingBean, DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(CuratorZkClient.class);

    private CuratorFramework    client;

    private LeaderLatch         leaderLatch;

    private String              connStr;

    private int                 connTimeout;

    @Override
    public void init() throws Exception {
        afterPropertiesSet();
    }

    @Override
    public void destroy() throws Exception {
        doClose();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
            .connectString(connStr)
            .retryPolicy(new ExponentialBackoffRetry(20000, 3))
            .connectionTimeoutMs(connTimeout);

        client = builder.build();

        client.getConnectionStateListenable().addListener(new ConnectionStateListener() {

            public void stateChanged(CuratorFramework client, ConnectionState state) {
                if (state == ConnectionState.LOST) {
                    CuratorZkClient.this.stateChanged(StateListener.DISCONNECTED);
                } else if (state == ConnectionState.CONNECTED) {
                    CuratorZkClient.this.stateChanged(StateListener.CONNECTED);
                } else if (state == ConnectionState.RECONNECTED) {
                    CuratorZkClient.this.stateChanged(StateListener.RECONNECTED);
                } else if (state == ConnectionState.SUSPENDED) {
                    CuratorZkClient.this.stateChanged(StateListener.DISCONNECTED);
                }
            }
        });

        client.start();

        // 选举leader
        leaderLatch = new LeaderLatch(client, ZkContants.LATCH);
        leaderLatch.addListener(new LeaderLatchListener() {

            @Override
            public void isLeader() {
                createEphemeral(ZkContants.LEADER_HOST, AddressUtils.getHostIp(), false);
                logger.info("current host: " + AddressUtils.getHostIp() + " is leader!");
            }

            @Override
            public void notLeader() {
            }
        });

        leaderLatch.start();

        if (!exists(ZkContants.SERVICES)) {
            createPersistent(ZkContants.SERVICES, false);
        }
    }

    public class PathChildrenListener {

        private PathChildrenCache         childrenCache;
        private PathChildrenCacheListener childrenCacheListener;
        private AtomicBoolean             start = new AtomicBoolean(false);

        public PathChildrenListener(String path, final ChildListener listener){
            childrenCache = new PathChildrenCache(client, path, true);
            childrenCacheListener = new PathChildrenCacheListener() {

                @Override
                public void childEvent(CuratorFramework c,
                                       org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent event)
                                                                                                               throws Exception {

                    switch (event.getType()) {
                        case CHILD_ADDED:
                        case CHILD_REMOVED:
                        case CHILD_UPDATED:
                            String childPath = event.getData().getPath();
                            String parentPath = childPath.substring(0, childPath.lastIndexOf("/"));
                            List<String> children = client.getChildren().forPath(parentPath);
                            listener.childChanged(parentPath, children);
                        default:
                            break;
                    }
                }
            };
        }

        public void startListener() {
            try {
                if (start.compareAndSet(false, true)) {
                    childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                    childrenCache.getListenable().addListener(childrenCacheListener);
                }
            } catch (Exception e) {
                throw new ZkException(e);
            }
        }

        public void stopListener() {
            try {
                if (start.compareAndSet(true, false)) {
                    childrenCache.getListenable().removeListener(childrenCacheListener);
                    childrenCache.clear();
                    childrenCache.close();
                }
            } catch (IOException e) {
                throw new ZkException(e);
            }
        }
    }

    public class NodeListener {

        private NodeCache         nodeCache;
        private NodeCacheListener nodeCacheListener;
        private AtomicBoolean     start = new AtomicBoolean(false);

        public NodeListener(String path, final DataListener listener){
            nodeCache = new NodeCache(client, path, false);
            nodeCacheListener = new NodeCacheListener() {

                @Override
                public void nodeChanged() throws Exception {
                    String path = nodeCache.getCurrentData().getPath();

                    Object data = nodeCache.getCurrentData().getData();

                    if (data == null) {
                        listener.dataDeleted(path);
                    } else {
                        listener.dataChange(path, data);
                    }
                }
            };
        }

        public void startListener() {
            try {
                if (start.compareAndSet(false, true)) {
                    nodeCache.start(true);
                    nodeCache.getListenable().addListener(nodeCacheListener);
                }
            } catch (Exception e) {
                throw new ZkException(e);
            }
        }

        public void stopListener() {
            try {
                if (start.compareAndSet(true, false)) {
                    nodeCache.getListenable().removeListener(nodeCacheListener);
                    nodeCache.close();
                }
            } catch (IOException e) {
                throw new ZkException(e);
            }
        }
    }

    @Override
    protected void doClose() {
        try {
            leaderLatch.close();
        } catch (IOException e) {
        }
        client.close();
    }

    @Override
    protected String createPersistent(String path, boolean sequential) {
        try {
            if (sequential) {
                return client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path);
            } else {
                return client.create().withMode(CreateMode.PERSISTENT).forPath(path);
            }
        } catch (KeeperException.NodeExistsException e) {
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createPersistent(String path, String data, boolean sequential) {
        try {
            if (sequential) {
                return client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path);
            } else {
                return client.create()
                    .withMode(CreateMode.PERSISTENT)
                    .forPath(path, data.getBytes(Charset.forName("UTF-8")));
            }
        } catch (KeeperException.NodeExistsException e) {
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createEphemeral(String path, boolean sequential) {
        try {
            if (sequential) {
                return client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
            } else {
                return client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
            }
        } catch (KeeperException.NodeExistsException e) {
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    protected String createEphemeral(String path, String data, boolean sequential) {
        try {
            if (sequential) {
                return client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
            } else {
                return client.create()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(path, data.getBytes(Charset.forName("UTF-8")));
            }
        } catch (KeeperException.NodeExistsException e) {
            return path;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    protected PathChildrenListener createTargetChildListener(final String path, final ChildListener listener) {
        return new PathChildrenListener(path, listener);
    }

    protected List<String> addTargetChildListener(String path, PathChildrenListener listener) {
        try {
            listener.startListener();
            return client.getChildren().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            return null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    protected void removeTargetChildListener(String path, PathChildrenListener listener) {
        listener.stopListener();
    }

    @Override
    protected void addTargetDataListener(String path, NodeListener listener) {
        listener.startListener();
    }

    @Override
    protected NodeListener createTargetDataListener(String path, final DataListener listener) {
        return new NodeListener(path, listener);
    }

    @Override
    protected void removeTargetDataListener(String path, NodeListener listener) {
        listener.stopListener();
    }

    @Override
    public boolean delete(String path) {
        try {
            client.delete().forPath(path);
            return true;
        } catch (KeeperException.NoNodeException e) {
            return true;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public boolean exists(String path) {
        try {
            return client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T getData(String path) {
        try {
            return (T) client.getData().forPath(path);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public void setData(String path, String data) {
        try {
            client.setData().forPath(path, data.getBytes(Charset.forName("UTF-8")));
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public List<String> getChildren(String path) {
        try {
            return client.getChildren().forPath(path);
        } catch (KeeperException.NoNodeException e) {
            return null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public boolean isConnected() {
        return client.getZookeeperClient().isConnected();
    }

    public String getConnStr() {
        return connStr;
    }

    public void setConnStr(String connStr) {
        this.connStr = connStr;
    }

    public int getConnTimeout() {
        return connTimeout;
    }

    public void setConnTimeout(int connTimeout) {
        this.connTimeout = connTimeout;
    }

    @Override
    public boolean isLeader() {
        try {
            return leaderLatch.getLeader().isLeader();
        } catch (Exception e) {
            return false;
        }
    }

}
